package com.cennavi.bigdata.baotou.scripts;

import com.alibaba.fastjson.JSONObject;
import com.cennavi.bigdata.baotou.util.PropertiesUtil;
import com.mongodb.*;
import com.navinfo.mapspotter.foundation.util.MercatorUtil;
import com.vividsolutions.jts.geom.Coordinate;
import com.vividsolutions.jts.geom.Envelope;
import com.vividsolutions.jts.geom.LineString;
import com.vividsolutions.jts.io.WKBReader;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;

import java.io.FileReader;
import java.math.BigInteger;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by zuoweiguang on 2016/9/19.
 * 脚本功能：将世纪高通提供的rtic路况匹配道路的文档：correspondingOfzhuhai.csv 转换 存储到mongo中
 * 1.将 Corresponding表录入到mongo，并匹配好rtic与link的对应关系
 * 2.生成traffic_status表
 */
public class Corresponding2Mongo_bak {

    private static JSONObject prop = PropertiesUtil.getProperties();
    private WKBReader wkbReader = new WKBReader();
    public SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmm");
    private MongoClient mongoClient = null;
    private DB db = null;
    private DBCollection correspondingCol = null;
    private DBCollection trafficStatusCol = null;

    public Connection pgConn;
    public Statement stmt = null;
    public Connection oraConn;
    public Statement oraStmt = null;

    public Corresponding2Mongo_bak() {
        //获取 oracle 的连接
        String oraHost = "192.168.4.166";
        int oraPort = 1521;
        String oraDb = "orcl";
        String oraUser = "gdb_16aut";
        String oraPwd = "zaq1";
        try {
            Class.forName("oracle.jdbc.OracleDriver");
            String oraUrl = "jdbc:oracle:thin:@" + oraHost + ":" + oraPort + ":" + oraDb;
            oraConn = DriverManager.getConnection(oraUrl, oraUser, oraPwd);
            oraStmt = oraConn.createStatement();
        } catch (Exception e) {
            e.printStackTrace();
        }

        //获取 pg 的连接
        String pgHost = prop.getString("pgHost");
        int pgPort = prop.getInteger("pgPort");
        String pgDb = prop.getString("pgDb");
        String pgUser = prop.getString("pgUser");
        String pgPwd = prop.getString("pgPwd");
        try {
            Class.forName("org.postgresql.Driver");
            String url = "jdbc:postgresql://"+pgHost+":"+pgPort+"/"+pgDb+"";
            pgConn = DriverManager.getConnection(url, pgUser, pgPwd);
            stmt = pgConn.createStatement();
        } catch (Exception e) {
            e.printStackTrace();
        }
        //获取 mongo 的连接
        ServerAddress serverAddress = new ServerAddress(prop.getString("mongoHost"), prop.getInteger("mongoPort"));
        String host = prop.getString("mongoHost");
        if (host.equals("192.168.4.128")) {
            mongoClient = new MongoClient(serverAddress);
        } else {
            List mongoCredentialList = new ArrayList();
            mongoCredentialList.add(MongoCredential.createCredential("sa", "admin", "mm@2016".toCharArray()));
            mongoClient = new MongoClient(serverAddress, mongoCredentialList);
        }
        db = mongoClient.getDB(prop.getString("mongoDb"));
        correspondingCol = db.getCollection("corresponding");
        correspondingCol.drop();
        trafficStatusCol = db.getCollection("traffic_status");
        trafficStatusCol.drop();
    }

    public void makeCorresponding(String filename) {
        FileReader fileReader = null;
        CSVParser csvFileParser = null;
        //创建CSVFormat（header mapping）
        CSVFormat csvFileFormat = CSVFormat.DEFAULT;
        try {
            //初始化FileReader object
            fileReader = new FileReader(filename);
            csvFileParser = new CSVParser(fileReader, csvFileFormat);
            //CSV文件records
            List<CSVRecord> csvRecords = csvFileParser.getRecords();
//            System.out.println("corresponding records total:" + csvRecords.size());
            List<DBObject> objList = new ArrayList<>();
            int count = 0;
            for (int i = 1; i < csvRecords.size(); i++) {
                CSVRecord line = csvRecords.get(i);
                int columeSize = line.size();
//                System.out.println("size:" + columeSize);

                DBObject lineObj = new BasicDBObject();
                List<Integer> linkPidList = new ArrayList<>();
                for (int j = 0; j < columeSize; j ++) {
                    if (j == 0) {
                        int MeshNo = Integer.valueOf(line.get(j));
                        lineObj.put("mesh_code", MeshNo);
                        int RticLinkKind = Integer.valueOf(line.get(j + 1));
                        lineObj.put("rtic_link_kind", RticLinkKind);
                        String RTIC_linkID = line.get(j + 2);
                        String rtic_id = RTIC_linkID.substring(6, RTIC_linkID.length());
                        lineObj.put("rtic_id", String.format("%s%s%s", MeshNo, RticLinkKind, rtic_id));
                    }
                    else if (j >= 6 && j % 2 == 0) {
                        if (null != line.get(j)) {
                            int NILink = Integer.valueOf(line.get(j));
                            linkPidList.add(NILink);
                        }
                    }
                }
                lineObj.put("link_pid_list", linkPidList);
                objList.add(lineObj);
                count += 1;
                if (count % 2000 == 0) {
                    System.out.println("convert mongo count:" + count);
                    //存mongo
                    correspondingCol.insert(objList);
                    objList.clear();
                }
//                System.out.println(lineObj.toString());
            }
            if (objList.size() > 0) {
                correspondingCol.insert(objList);
                objList.clear();
            }
            System.out.println("convert mongo count:" + count);

            //创建索引
            correspondingCol.createIndex("mesh_code");
            correspondingCol.createIndex("rtic_id");
            correspondingCol.createIndex("rtic_link_kind");
            correspondingCol.createIndex("link_pid_list");
            System.out.println("make corresponding index");

            System.out.println("------------- make corresponding done! --------------");
        } catch (Exception e) {
            e.printStackTrace();
            try {
                fileReader.close();
                csvFileParser.close();
            } catch (Exception e1) {
                e1.printStackTrace();
            }
        } finally {
            try {
                fileReader.close();
                csvFileParser.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public void makeTrafficStatus() {
        Map<Integer, Integer> linkCount = new HashMap();
        List<String> linkList = new ArrayList();
        int rticCount = 0;
        try {
            DBCursor cursor = correspondingCol.find();
            while (cursor.hasNext()) {
                DBObject correspondingObj = cursor.next();
                Integer mesh_code = (Integer) correspondingObj.get("mesh_code");
                String rtic_id = (String) correspondingObj.get("rtic_id");
                Integer rtic_link_kind = (Integer) correspondingObj.get("rtic_link_kind");
                List<Integer> link_pid_list = (List<Integer>) correspondingObj.get("link_pid_list");

                if (link_pid_list.size() == 0) {
                    continue;
                }

                //-------------------------------- 到 pg 中查询对应的Linkid --------------------------------------------
                for (int link_pid : link_pid_list) {
                    if (!linkList.contains(mesh_code + "," + rtic_id + "," + rtic_link_kind + "," + link_pid)) {
                        linkList.add(mesh_code + "," + rtic_id + "," + rtic_link_kind + "," + link_pid);
                    } else {
//                        System.out.println(mesh_code + "," + rtic_id + "," + rtic_link_kind + "," + link_pid);
                    }
                }

                StringBuffer sb = new StringBuffer();
                for (int link_pid: link_pid_list) {
                    BigInteger b = new BigInteger(String.valueOf(link_pid));
                    sb.append(b.abs() + ",");
                }
                String pidStrList = sb.substring(0, sb.length() - 1);
                //查询 pg 对应的道路
                String sql = "SELECT a.pid, a.direct, a.functionclass, st_asewkb(a.geom) as geom, b.name " +
                        "FROM road a LEFT JOIN road_name b ON a.pid=b.link_pid " +
                        "WHERE a.pid in (" + pidStrList.toString() + ")";
                ResultSet rs = stmt.executeQuery(sql);
                //-------------------------------- 到 pg 中查询对应的Linkid --------------------------------------------
                boolean rsNull = true;
                while (rs.next()) {
                    try {
                        DBObject trafficObj = new BasicDBObject();
                        int link_id = rs.getInt("pid");
                        trafficObj.put("mesh_code", mesh_code);
                        trafficObj.put("link_id", link_id);
                        trafficObj.put("link_name", rs.getString("name"));
                        trafficObj.put("direct", rs.getInt("direct"));
                        trafficObj.put("rtic_link_kind", rtic_link_kind);
                        trafficObj.put("functionclass", rs.getInt("functionclass"));
                        trafficObj.put("status", 1);
                        trafficObj.put("temp_status", 1);
                        trafficObj.put("rtic_id", rtic_id);
                        trafficObj.put("traveltime", 0);
                        trafficObj.put("updatetime", sdf.format(new java.util.Date()));
                        trafficObj.put("adcode", "");
                        trafficObj.put("rtic_updatetime", "");
                        trafficObj.put("version", "");
                        trafficObj.put("sectionCntList", new ArrayList<>());
                        byte[] wkb = rs.getBytes("geom");
                        LineString lineString = (LineString) wkbReader.read(wkb);
                        trafficObj.put("geometry", lineString.toString());
                        Envelope envelope = lineString.getEnvelope().getEnvelopeInternal();
                        //生成tile
                        Coordinate lt = new Coordinate(envelope.getMinX(), envelope.getMaxY());
                        Coordinate rb = new Coordinate(envelope.getMaxX(), envelope.getMinY());
                        //tile级别
                        for (int k = 7; k <= 17; k++) {
                            trafficObj.put("tile_" + k, MercatorUtil.bounds2Tiles(lt, rb, k));
                        }
                        BasicDBObject query = new BasicDBObject();
                        query.append("mesh_code", mesh_code);
                        query.append("link_id", link_id);
                        query.append("rtic_link_kind", rtic_link_kind);
                        query.append("rtic_id", rtic_id);
                        DBObject update = new BasicDBObject("$set", trafficObj);
                        trafficStatusCol.update(query, update, true, false);
                        linkCount.put(link_id, 1);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    rsNull = false;
                }
                if (rsNull) {
                    //到 oracle 中查询漏掉的road
                    String oraSql = "SELECT l.LINK_PID as pid, l.DIRECT as direct, l.FUNCTION_CLASS as function_class, " +
                            "l.KIND as kind, l.GEOMETRY.get_wkb() as geom FROM rd_link l where l.LINK_PID in (" + pidStrList.toString() + ")";
                    ResultSet oraRs = oraStmt.executeQuery(oraSql);
                    boolean isNull = true;
                    while (oraRs.next()) {
                        try {
                            DBObject trafficObj = new BasicDBObject();
                            int link_id = oraRs.getInt("pid");
                            trafficObj.put("mesh_code", mesh_code);
                            trafficObj.put("link_id", link_id);
                            trafficObj.put("link_name", "");
                            trafficObj.put("direct", oraRs.getInt("direct"));
                            trafficObj.put("rtic_link_kind", rtic_link_kind);
                            trafficObj.put("functionclass", oraRs.getInt("function_class"));
                            trafficObj.put("status", 1);
                            trafficObj.put("temp_status", 1);
                            trafficObj.put("rtic_id", rtic_id);
                            trafficObj.put("traveltime", 0);
                            trafficObj.put("updatetime", sdf.format(new java.util.Date()));
                            trafficObj.put("adcode", "");
                            trafficObj.put("rtic_updatetime", "");
                            trafficObj.put("version", "");
                            trafficObj.put("sectionCntList", new ArrayList<>());
                            byte[] wkb = oraRs.getBytes("geom");
                            LineString lineString = (LineString) wkbReader.read(wkb);
                            trafficObj.put("geometry", lineString.toString());
                            Envelope envelope = lineString.getEnvelope().getEnvelopeInternal();
                            //生成tile
                            Coordinate lt = new Coordinate(envelope.getMinX(), envelope.getMaxY());
                            Coordinate rb = new Coordinate(envelope.getMaxX(), envelope.getMinY());
                            //tile级别
                            for (int k = 7; k <= 17; k++) {
                                trafficObj.put("tile_" + k, MercatorUtil.bounds2Tiles(lt, rb, k));
                            }
                            BasicDBObject query = new BasicDBObject();
                            query.append("mesh_code", mesh_code);
                            query.append("link_id", link_id);
                            query.append("rtic_link_kind", rtic_link_kind);
                            query.append("rtic_id", rtic_id);
                            DBObject update = new BasicDBObject("$set", trafficObj);
                            trafficStatusCol.update(query, update, true, false);
                            linkCount.put(link_id, 1);
                            isNull = false;
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    if (isNull) {
//                        System.out.println(pidStrList.toString());
                    }
                }
                rticCount += 1;
                if (rticCount % 2000 == 0) {
                    System.out.println("make rtic count:" + rticCount);
                }
            }
            //创建索引
            trafficStatusCol.createIndex("mesh_code");
            trafficStatusCol.createIndex("link_id");
            trafficStatusCol.createIndex("rtic_link_kind");
            trafficStatusCol.createIndex("functionclass");
            trafficStatusCol.createIndex("rtic_id");
            trafficStatusCol.createIndex("mesh_code");
            for (int k = 10; k <= 17; k++) {
                trafficStatusCol.createIndex("tile_" + k);
            }
            System.out.println("make traffic_status index");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            System.out.println("make rtic count:" + rticCount);
            System.out.println("make link count:" + linkCount.keySet().size());
            System.out.println("-------------- make traffic_status done! --------------");
        }
    }

    private void closeConnection() {
        try {
            if (pgConn != null) {
                pgConn.close();
            }
            if (oraConn != null) {
                oraConn.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        if (mongoClient != null) {
            mongoClient.close();
        }
    }

    public static void main(String[] args) {
        Corresponding2Mongo_bak ccm = new Corresponding2Mongo_bak();
        ccm.makeCorresponding("correspondingOfbaotou_160927.csv");
        ccm.makeTrafficStatus();
        ccm.closeConnection();
    }

}
